package nil.ed.simulator.request.scheduler.impl;

import lombok.Getter;
import nil.ed.simulator.request.scheduler.ScheduleUnit;
import nil.ed.simulator.request.scheduler.Scheduler;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;

import java.io.IOException;
import java.util.Comparator;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 优先级调度器.
 */
@Getter
public class PriorityScheduler implements Scheduler {
    private final PriorityBlockingQueue<PriorityScheduleUnit> scheduleUnits;
    private final CycleThreadPoolExecutor executorService;
    private final String bizGroup;
    private final int concurrency;

    @SuppressWarnings({"rawtypes", "unchecked"})
    public PriorityScheduler(String bizGroup, int concurrency) {
        this.bizGroup = bizGroup;
        this.concurrency = concurrency;
        this.scheduleUnits = new PriorityBlockingQueue<>(100, Comparator.comparingInt(PriorityScheduleUnit::getPriority));
        this.executorService = new CycleThreadPoolExecutor(concurrency, concurrency,
                10, TimeUnit.SECONDS, (BlockingQueue) scheduleUnits, new BasicThreadFactory.Builder().namingPattern("Scheduler-" + bizGroup + "-%s").build(), new ThreadPoolExecutor.AbortPolicy());
        CountDownLatch latch = new CountDownLatch(concurrency);
        this.executorService.execute(latch::countDown);
    }

    @Override
    public void addScheduleUnit(ScheduleUnit scheduleUnit) {
        if (!(scheduleUnit instanceof PriorityScheduleUnit)) {
            throw new IllegalArgumentException("Incompatible type: " + scheduleUnit.getClass().getSimpleName());
        }
        this.scheduleUnits.add((PriorityScheduleUnit) scheduleUnit);
    }

    @Override
    public Set<ScheduleUnit> getSchedulingTasks() {
        return executorService.getSchedulingTasks().stream().map(ScheduleUnit.class::cast).collect(Collectors.toSet());
    }

    @Override
    public int getWaitingCount() {
        return scheduleUnits.size();
    }

    @Override
    public void close() throws IOException {
        this.executorService.shutdown();
    }

    public static void main(String[] args) {
        AtomicInteger integer = new AtomicInteger();
        AtomicInteger priority = new AtomicInteger(Integer.MAX_VALUE);
        Scheduler scheduler = new PriorityScheduler("Test", 3);
        scheduler.addScheduleUnit(new PriorityScheduleUnit() {
            @Override
            public int getPriority() {
                return priority.getAndDecrement();
            }

            @Override
            public boolean isTerminate() {
                return integer.get() == 10;
            }

            @Override
            public String getBizGroup() {
                return "Test";
            }

            @Override
            public void run() {
                System.out.println("OKKK" + integer.getAndIncrement());
            }
        });
    }
}
